// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Support DServer-DClient communication
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: change idl check from md5 to crc16
// ------------------------------------------------------------------

#include <discovery/endpoint/EDPSimpleListeners.h>

#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <discovery/endpoint/EDPSimple.h>
#include <discovery/participant/PDPSimple.h>

#include <edds/rtps/common/InstanceHandle.h>

#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>

#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/writer/StatefulWriter.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <mutex>

#include <transport/TransportFactory.hpp>
#include <elog/Log.hpp>
#include <edds/dds/core/policy/ParameterList.hpp>

using ParameterList = evbs::edds::dds::ParameterList;

// Release reader lock to avoid ABBA lock. PDP mutex should always be first.
// Keep change information on local variables to check consistency later
#define PREVENT_PDP_DEADLOCK(reader, change, pdp)                                                                      \
    GUID_t writer_guid = (change)->writerGUID;                                                                         \
    SequenceNumber_t seq_num = (change)->sequenceNumber;                                                               \
    (reader)->getMutex().unlock();                                                                                     \
    std::unique_lock<std::recursive_mutex> lock(*((pdp)->getMutex()));                                                 \
    (reader)->getMutex().lock();                                                                                       \
                                                                                                                       \
    if ((ALIVE != (change)->kind) || (seq_num != (change)->sequenceNumber) || (writer_guid != (change)->writerGUID)) { \
        return;                                                                                                        \
    }                                                                                                                  \
    (void)seq_num

namespace evbs {
namespace ertps {
namespace rtps {

void EDPBasePUBListener::add_writer_from_change(RTPSReader* reader, ReaderHistory* reader_history,
                                                CacheChange_t* change, EDP* edp, bool release_change /*=true*/,
                                                const WriterAddedCallback& writer_added_callback /* = nullptr*/) {
    // LOAD INFORMATION IN DESTINATION WRITER PROXY DATA
    CDRMessage_t tempMsg(change->serializedPayload);
    reader->getMutex().unlock();
    auto temp_writer_data = edp->get_temporary_writer_proxies_pool().get();
    reader->getMutex().lock();

    if (temp_writer_data->readFromCDRMessage(&tempMsg, edp->mp_RTPSParticipant)) {
        if (temp_writer_data->guid().guidPrefix == edp->mp_RTPSParticipant->getGuid().guidPrefix) {
            return;
        }
        bool take_locator = false;
        // LOAD INFORMATION IN DESTINATION WRITER PROXY DATA
        auto copy_data_fun = [&temp_writer_data, &take_locator](WriterProxyData* data, bool updating,
                                                                const ParticipantProxyData& participant_data,
                                                                RTPSParticipantImpl* localparticipant) {
            if (!temp_writer_data->has_locators()) {
                if (participant_data.m_VendorId != c_VendorId_vbs) {
                    temp_writer_data->set_locators(participant_data.default_locators);
                } else {
                    temp_writer_data->set_remote_locators(participant_data.default_locators, localparticipant, true);
                }
            } else {
                take_locator = temp_writer_data->has_udp_locators();
            }

            if (updating && (!data->is_update_allowed(*temp_writer_data))) {
                logWarning(RTPS_EDP, "Received incompatible update for WriterQos. writer_guid = " << data->guid());
            }
            *data = *temp_writer_data;
            data->setIdlCrc16((temp_writer_data->getIdlCrc16()).c_str());
            if (participant_data.m_guid.guidPrefix.is_guid_static() && participant_data.m_hostId != 0) {
                data->set_host_id(participant_data.m_hostId);
                data->set_pid(participant_data.m_pid);
            } else {
                data->set_host_id(participant_data.m_guid.guidPrefix.get_host_id());
                data->set_pid(participant_data.m_guid.guidPrefix.get_process_id());
            }
            return true;
        };

        GUID_t participant_guid;
        WriterProxyData* writer_data =
            edp->mp_PDP->addWriterProxyData(temp_writer_data->guid(), participant_guid, copy_data_fun);

        // release temporary proxy
        temp_writer_data.reset();

        // Removing change from history
        (void)reader_history->remove_change(change, release_change);

        // At this point we can release reader lock, cause change is not used
        reader->getMutex().unlock();
        if (writer_data != nullptr) {
            (void)edp->pairing_writer_proxy_with_any_local_reader(participant_guid, writer_data);
            if (nullptr != writer_added_callback) {
                writer_added_callback(reader, change, writer_data->m_qos, take_locator);
            }
        } else  // NOT ADDED BECAUSE IT WAS ALREADY THERE
        {
            logWarning(RTPS_EDP, "Received message from UNKNOWN RTPSParticipant, removing");
        }
        // Take again the reader lock.
        reader->getMutex().lock();
    }
}
void EDPSimplePUBListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                 const CacheChange_t* const change_in) {
    CacheChange_t* change = (CacheChange_t*)change_in;
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    // std::lock_guard<std::recursive_mutex>
    // guard(*this->sedp_->publications_reader_.first->getMutex());
    if (!computeKey(change)) {
        logWarning(RTPS_EDP, "Received change with no Key");
    }

    ReaderHistory* reader_history =
#if HAVE_SECURITY
        reader == sedp_->publications_secure_reader_.first ? sedp_->publications_secure_reader_.second :
#endif  // if HAVE_SECURITY
                                                           sedp_->publications_reader_.second;

    if (change->kind == ALIVE) {
        PREVENT_PDP_DEADLOCK(reader, change, sedp_->mp_PDP);

        // Note: change is removed from history inside this method.
        add_writer_from_change(reader, reader_history, change, sedp_);
    } else {
        GUID_t writer_guid = vbsutil::xmlparser::iHandle2GUID(change->instanceHandle);
        // Removing change from history
        (void)reader_history->remove_change(change);
        reader->getMutex().unlock();
        (void)this->sedp_->mp_PDP->removeWriterProxyData(writer_guid);
        reader->getMutex().lock();
    }
}

bool EDPListener::computeKey(CacheChange_t* change) {
    return ParameterList::readInstanceHandleFromCDRMsg(change, edds::dds::PID_ENDPOINT_GUID);
}

void EDPBaseSUBListener::add_reader_from_change(RTPSReader* reader, ReaderHistory* reader_history,
                                                CacheChange_t* change, EDP* edp, bool release_change /*=true*/,
                                                const ReaderAddedCallback& reader_added_callback /* = nullptr*/) {
    // LOAD INFORMATION IN TEMPORAL WRITER PROXY DATA
    CDRMessage_t tempMsg(change->serializedPayload);
    reader->getMutex().unlock();
    auto temp_reader_data = edp->get_temporary_reader_proxies_pool().get();
    reader->getMutex().lock();

    if (temp_reader_data->readFromCDRMessage(&tempMsg, edp->mp_RTPSParticipant)) {
        if (temp_reader_data->guid().guidPrefix == edp->mp_RTPSParticipant->getGuid().guidPrefix) {
            return;
        }
        bool take_locator = false;
        uint32_t queue_id = 0;
        auto copy_data_fun = [&temp_reader_data, &take_locator, &queue_id](ReaderProxyData* data, bool updating,
                                                                           const ParticipantProxyData& participant_data,
                                                                           RTPSParticipantImpl* localparticipant) {
            if (!temp_reader_data->has_locators()) {
                if (participant_data.m_VendorId != c_VendorId_vbs) {
                    temp_reader_data->set_locators(participant_data.default_locators);
                } else {
                    temp_reader_data->set_remote_locators(participant_data.default_locators, localparticipant, true);
                }
            } else {
                // Confirm whether it carries UDP locator to determine whether to forward it to mvbs
                // Confirm whether the dsf locator is carried and complete the domain authentication
                temp_reader_data->has_udp_dsf_locators(take_locator, queue_id);
            }
            if (updating && (!data->is_update_allowed(*temp_reader_data))) {
                logWarning(RTPS_EDP, "Received incompatible update for ReaderQos. reader_guid = " << data->guid());
            }
            *data = *temp_reader_data;
            data->setIdlCrc16((temp_reader_data->getIdlCrc16()).c_str());

            if (participant_data.m_guid.guidPrefix.is_guid_static() && participant_data.m_hostId != 0) {
                data->set_host_id(participant_data.m_hostId);
                data->set_pid(participant_data.m_pid);
            } else {
                data->set_host_id(participant_data.m_guid.guidPrefix.get_host_id());
                data->set_pid(participant_data.m_guid.guidPrefix.get_process_id());
            }
            return true;
        };

        // LOOK IF IS AN UPDATED INFORMATION
        GUID_t participant_guid;
        ReaderProxyData* reader_data =
            edp->mp_PDP->addReaderProxyData(temp_reader_data->guid(), participant_guid, std::move(copy_data_fun));

        // Release the temporary proxy
        temp_reader_data.reset();

        // Remove change from history.
        (void)reader_history->remove_change(change, release_change);

        // At this point we can release reader lock, cause change is not used
        reader->getMutex().unlock();

        if (reader_data != nullptr)  // ADDED NEW DATA
        {
            (void)edp->pairing_reader_proxy_with_any_local_writer(participant_guid, reader_data);
            if (nullptr != reader_added_callback) {
                reader_added_callback(reader, change, reader_data->m_qos, take_locator, queue_id);
            }
        } else {
            logWarning(RTPS_EDP, "From UNKNOWN RTPSParticipant, removing");
        }

        // Take again the reader lock.
        reader->getMutex().lock();
    }
}

void EDPSimpleSUBListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                 const CacheChange_t* const change_in) {
    CacheChange_t* change = (CacheChange_t*)change_in;
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    // std::lock_guard<std::recursive_mutex>
    // guard(*this->sedp_->subscriptions_reader_.first->getMutex());
    if (!computeKey(change)) {
        logWarning(RTPS_EDP, "Received change with no Key");
    }

    ReaderHistory* reader_history =
#if HAVE_SECURITY
        reader == sedp_->subscriptions_secure_reader_.first ? sedp_->subscriptions_secure_reader_.second :
#endif  // if HAVE_SECURITY
                                                            sedp_->subscriptions_reader_.second;

    if (change->kind == ALIVE) {
        PREVENT_PDP_DEADLOCK(reader, change, sedp_->mp_PDP);

        // Note: change is removed from history inside this method.
        add_reader_from_change(reader, reader_history, change, sedp_);
    } else {
        GUID_t reader_guid = vbsutil::xmlparser::iHandle2GUID(change->instanceHandle);
        // Removing change from history
        (void)reader_history->remove_change(change);
        reader->getMutex().unlock();
        (void)this->sedp_->mp_PDP->removeReaderProxyData(reader_guid);
        reader->getMutex().lock();
    }
}

void EDPSimplePUBListener::onWriterChangeReceivedByAll(vbs::dispatcher::SubstrateWriter* const writer,
                                                       CacheChange_t* change) {
    (void)writer;
    if (ChangeKind_t::NOT_ALIVE_DISPOSED_UNREGISTERED == change->kind) {
        WriterHistory* writer_history =
#if HAVE_SECURITY
            writer == sedp_->publications_secure_writer_.first ? sedp_->publications_secure_writer_.second :
#endif  // if HAVE_SECURITY
                                                               sedp_->publications_writer_.second;

        (void)writer_history->remove_change(change);
    }
}

void EDPSimpleSUBListener::onWriterChangeReceivedByAll(vbs::dispatcher::SubstrateWriter* const writer,
                                                       CacheChange_t* change) {
    (void)writer;
    if (ChangeKind_t::NOT_ALIVE_DISPOSED_UNREGISTERED == change->kind) {
        WriterHistory* writer_history =
#if HAVE_SECURITY
            writer == sedp_->subscriptions_secure_writer_.first ? sedp_->subscriptions_secure_writer_.second :
#endif  // if HAVE_SECURITY
                                                                sedp_->subscriptions_writer_.second;

        (void)writer_history->remove_change(change);
    }
}

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
